#pragma once

#include<iostream>
#include<unordered_map>
#include<vector>
#include<string>
#include<mutex>
#include<condition_variable>
#include<thread>
#include<utility>
#include<typeinfo>
#include<functional>
#include<algorithm>
#include<cstring>
#include<ctime>
#include<cassert>
#include<cstdlib>
#include<cerrno>
#include<unistd.h>
#include<fcntl.h>
#include<sys/timerfd.h>
#include<sys/eventfd.h>
#include<sys/types.h>
#include<sys/socket.h>
#include<sys/epoll.h>
#include<netinet/in.h>
#include<arpa/inet.h>

#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL INF

#define LOG(level,format,...) do\
{\
      if(level<LOG_LEVEL)     break;\
      time_t t=time(nullptr);\
      struct tm* lt=localtime(&t);\
      char tmp[32]={0};\
      strftime(tmp,31,"%H:%M:%S",lt);\
      fprintf(stdout,"[%s %s:%d]" format "\n",tmp,__FILE__,__LINE__,##__VA_ARGS__);\
}while(0)

#define INF_LOG(format,...) LOG(INF,format,##__VA_ARGS__)
#define DBG_LOG(format,...) LOG(DBG,format,##__VA_ARGS__)
#define ERR_LOG(format,...) LOG(ERR,format,##__VA_ARGS__)


#define BUFFER_DEFAULT_SIZE 1024
#define MAX_BACKLOG 10

class Buffer
{
public:
      Buffer():buffer_(BUFFER_DEFAULT_SIZE),reader_index_(0),writer_index_(0){}
      
      ~Buffer(){}
public:
      char* Begin(){return &*buffer_.begin();}
      //获取当前写入起始地址，_buffer的空间地址，加上写偏移量
      char* WriterPosition(){return Begin()+writer_index_;}
      //获取当前读入起始地址
      char* ReaderPosition(){return Begin()+reader_index_;}
      //获取缓冲区末尾空闲空间大小--写偏移之后的空闲空间，总体空间-写偏移
      uint64_t TailIdleSize(){return buffer_.size()-writer_index_;}
      //获取缓冲区起始空间大小--读之前的空闲空间
      uint64_t HeadIdleSize(){return reader_index_;}
      //获取可读数据大小
      uint64_t ReadAbleSize(){return writer_index_-reader_index_;}
      //将读偏移向后移动
      void MoveReadOffset(uint64_t len)
      {
            if(len==0)  return;
            assert(len<=ReadAbleSize());
            reader_index_+=len;
      }
      //将写偏移向后移动
      void MoveWriteOffset(uint64_t len)
      {
            if(len==0)  return;
            assert(len<=TailIdleSize());
            writer_index_+=len;
      }
      //确保可写空间足够（整体空间够了，就移动数据）不够就扩容
      void EnsureWriteSpace(uint64_t len)
      {
            //如果末尾空间足够，就直接返回
            if(TailIdleSize()>=len) return;
            //如果整体空间足够，就移动数据
            if(TailIdleSize()+HeadIdleSize()>=len)
            {
                  size_t sz=ReadAbleSize();//保存可读数据大小
                  std::copy(buffer_.begin(),buffer_.begin()+sz,WriterPosition());
                  reader_index_=0;
                  writer_index_=sz;
            }
            //不够，就扩容
            else 
            {
                  buffer_.resize(writer_index_+len); 
            }
      }
      //写入数据
      void Write(const void* data,uint64_t len)
      {
            if(len==0)  return;
            EnsureWriteSpace(len);
            const char* start=(const char*)data;
            std::copy(start,start+len,WriterPosition());  
      }
      void WriteAndPush(const void* data,uint64_t len)
      {
            Write(data,len);
            MoveWriteOffset(len);
      }
      void WriteString(const std::string& data)
      {
            Write(data.c_str(),data.size());
      }
      void WriteStringAndPush(const std::string& data)
      {
            WriteString(data);
            MoveWriteOffset(data.size());
      }
      void WriteBuffer(Buffer& buffer)
      {
            Write(buffer.ReaderPosition(),buffer.ReadAbleSize());
      }
      void WriteBufferAndPush(Buffer& buffer)
      {
            //std::cout<<"===="<<buffer.ReaderPosition()<<std::endl;;
            WriteBuffer(buffer);
            MoveWriteOffset(buffer.ReadAbleSize());
      }
      //读取数据
      void Read(void* buf,uint64_t len)
      {
            if(len==0)  return;
            assert(len<=ReadAbleSize());
            std::copy(ReaderPosition(),ReaderPosition()+len,(char*)buf);
      }
      void ReadPop(void* buf,uint64_t len)
      {
            Read(buf,len);
            MoveReadOffset(len);
      }
      std::string ReadAsString(uint64_t len)
      {
            assert(len<=ReadAbleSize());
            std::string s;
            s.resize(len);
            Read(&s[0],len);
            return s;
      }
      std::string ReadAsStringPop(uint64_t len)
      {
            std::string s=ReadAsString(len);
            MoveReadOffset(len);
            return s;
      }
      char* FindCRLf()
      {
            char* pos=(char*)memchr(ReaderPosition(),'\n',ReadAbleSize());
            return pos;
      }
      std::string GetLine()
      {
            char* pos=FindCRLf();
            if(pos==NULL)
                  return "";
            return ReadAsString(pos-ReaderPosition()+1);
      }
      std::string GetLineAndPop()
      {
            std::string s=GetLine();
            MoveReadOffset(s.length());
            return s;
      }
      //清空缓冲区
      void Clear()
      {
            reader_index_=0;
            writer_index_=0;
      }
private:
      std::vector<char> buffer_;//使用vector进行内存空间管理
      uint64_t reader_index_;   //读偏移
      uint64_t writer_index_;   //写偏移
};

class Socket
{
public:
      Socket():sockfd_(-1){}
      Socket(int fd):sockfd_(fd){}
      ~Socket(){}
public:
      //创建套接字
      bool Sock()
      {
            //int socket(int domain, int type, int protocol);
            sockfd_=socket(AF_INET,SOCK_STREAM,0);
            if(sockfd_==-1)
            {
                  DBG_LOG("sockfd create failed!!!!");
                  return false;
            }
            return true;
      }
      bool Bind(const std::string& ip,uint16_t port)
      {
            struct sockaddr_in addr;
            addr.sin_family=AF_INET;
            addr.sin_port=htons(port);
            addr.sin_addr.s_addr=inet_addr(ip.c_str());
            socklen_t len=sizeof(sockaddr_in);

            int n=bind(sockfd_,(const sockaddr*)&addr,len);
            if(n<0)
            {
                  DBG_LOG("bind fail");
                  return false;
            }
            return true;
      }
      bool Listen(int backlog=MAX_BACKLOG)
      {
            int n=listen(sockfd_,backlog);
            if(n<0)
            {
                  DBG_LOG("listen fail!!!");
                  return false;
            }
            return true;
      }
      //向服务器发起连接
      bool Connect(const std::string& ip,uint16_t port)
      {
            struct sockaddr_in addr;
            addr.sin_family=AF_INET;
            addr.sin_port=htons(port);
            addr.sin_addr.s_addr=inet_addr(ip.c_str());
            socklen_t len=sizeof(addr);

            int n=connect(sockfd_,(const sockaddr*)&addr,len);
            std::cout<<"向服务器发起连接:fd:"<<sockfd_<<std::endl;
            if(n<0)
            {
                  DBG_LOG("connect fail!!!!");
                  return false;
            }
            return true;
      }
      int Accept()
      {
            struct sockaddr_in addr;
            socklen_t len;
            int newfd=accept(sockfd_,(sockaddr*)&addr,&len);
            if(newfd<0)
            {
                  DBG_LOG("accept fail!!!");
                  return -1;
            }
            return newfd;
      }
      ssize_t Recv(void* buf,size_t len,int flag=0)
      {
            ssize_t n=recv(sockfd_,buf,len,flag);
            if(n<=0)
            {
                  //EAGAIN:当前socket的接受缓冲区没有数据了，在非阻塞的情况下菜会有这种情况
                  //EINTE:当前socket的阻塞等待，被信号打断了
                  if(errno==EAGAIN||errno==EINTR)
                        return 0;//表示这次没有接受到数据
                  DBG_LOG("recv fail!!!");
                  return -1;
            }
            return n;
      }
      ssize_t NonBlockRecv(void* buf,size_t len)
      {
            return Recv(buf,len,MSG_DONTWAIT);//MSG_DONTWAIT表示非阻塞
      }
      ssize_t Send(const void* buf,size_t len,int flag=0)
      {
            ssize_t n=send(sockfd_,buf,len,flag);
            if(n<=0)
            {
                  if(errno==EAGAIN||errno==EINTR)
                        return 0;
                  DBG_LOG("send fail!!!");
                  return -1;
            }
            return n;
      }
      ssize_t NonBlockSend(const void* buf,size_t len)
      {
            return Send(buf,len,MSG_DONTWAIT);
      }
      void Close()
      {
            if(sockfd_!=-1)
            {
                  close(sockfd_);
                  sockfd_=-1;
            }
      }
      int GetFd()
      {
            return sockfd_;
      }
      //创建一个服务端连接
      bool CreateServer(uint16_t port,const std::string ip="0.0.0.0",bool block_flag=false)
      {
            //1.创建一个套接字 2.设置非阻塞 3.绑定地址 4.开始监听 5.启动地址复用
            if(Sock()==false) return false;
            if(block_flag)    NonBlock();
            if(Bind(ip,port)==false) return false;
            if(Listen()==false)     return false;
            ReuseAddr();
            return true;
      }
      //创建一个客服端连接
      bool CreateClient(uint16_t port,const std::string ip)
      {
            if(Sock()==false) return false;
            if(Connect(ip,port)==false)   return false;
            return true;
      }
      void NonBlock()
      {
            int flag=fcntl(sockfd_,F_GETFL,0);
            fcntl(sockfd_,F_SETFL,flag|O_NONBLOCK);
      }
      void ReuseAddr()
      {
            //int setsockopt(int sockfd, int level, int optname,void *optval, socklen_t optlen);
            int optval=1;
            setsockopt(sockfd_,SOL_SOCKET,SO_REUSEADDR,(void*)&optval,sizeof(int));
            optval=1;
            setsockopt(sockfd_,SOL_SOCKET,SO_REUSEPORT,(void*)&optval,sizeof(int));
      }
private:
      int sockfd_;
};

class EventLoop;
class Poller;
class Channel
{
      using EventCallback=std::function<void()>;
public:
      Channel(int fd,EventLoop* loop):fd_(fd),loop_(loop),events_(0),revents_(0){}
public:
      int GetFd(){return fd_;}
      uint32_t Events(){return events_;}//获取想要被监控的事件
      void SetREvents(uint32_t revents){revents_=revents;}//设置实际就绪的事件
      void SetReadCallback(const EventCallback& cb){read_callback_=cb;}
      void SetWriteCallback(const EventCallback& cb){write_callback_=cb;}
      void SetErrorCallback(const EventCallback& cb){error_callback_=cb;}
      void SetCloseCallback(const EventCallback& cb){close_callback_=cb;}
      void SetEventCallback(const EventCallback& cb){event_callback_=cb;}
      //当前是否监控了可读
      bool ReadAble(){return (events_&EPOLLIN);}
      //当前是否监控了可写
      bool WriteAble(){return (events_&EPOLLOUT);}
      //启动了读事件监控
      void EnableRead(){events_|=EPOLLIN;Update();}
      //启动了写事件监控
      void EnableWrite(){events_|=EPOLLOUT;Update();}
      //关闭读事件监控
      void DisableRead(){events_&=~EPOLLIN;Update();}
      //关闭写事件监控
      void DisableWrite(){events_&=~EPOLLOUT;Update();}
      //事件处理,一旦连接触发了什么事件，就调用这个函数，自己触发了什么事件如何处理自己知道
      void Remove();
      void Update();
      void HandleEvent()
      {
            if((events_&EPOLLIN)||(events_&EPOLLHUP)||(events_&EPOLLPRI))
            {
                  if(read_callback_) read_callback_();
            }
            /*有可能会释放连接的操作事件，一次只处理一个*/
            if(events_&EPOLLOUT)
            {
                  if(write_callback_)     write_callback_();
            }
            else if(events_&EPOLLERR)
            {
                  if(error_callback_)     error_callback_();
            }
            else if(events_&EPOLLHUP)
            {
                  if(close_callback_)     close_callback_();
            }
            
            if(event_callback_)     event_callback_();
            //std::cout<<"hfdhfhd"<<std::endl;

      }
private:
      int fd_;
      EventLoop* loop_;
      uint32_t events_;//当前需要监控的事件
      uint32_t revents_;//当前连接触发的事件
      EventCallback read_callback_;   //可读事件被触发的回调函数    
      EventCallback write_callback_;  //可写事件被触发的回调函数
      EventCallback error_callback_;  //错误事件被触发的回调函数
      EventCallback close_callback_;  //关闭事件被触发的回调函数
      EventCallback event_callback_;  //任意事件被触发的回调函数
};

#define MAX_EPOLLEVENTS 1024
class Poller
{
private:
      //对epoll直接操作
      void Update(Channel* channel,int op)
      {
            int fd=channel->GetFd();
            struct epoll_event ev;
            ev.data.fd=fd;
            ev.events=channel->Events();
            int n=epoll_ctl(epfd_,op,fd,&ev);
            if(n<0)
            {
                  ERR_LOG("epoll_ctl fail!!");
            }
      }
      //判断一个Channel是否已经添加到了事件监控
      bool HasChannel(Channel* channel)
      {
            auto it=channels_.find(channel->GetFd());
            if(it==channels_.end())
            {
                  return false;
            }
            return true;
      }
public:
      Poller()
      {
            epfd_=epoll_create(MAX_EPOLLEVENTS);
            if(epfd_<0)
            {
                  ERR_LOG("epoll_create fail!!");
                  abort();
            }
      }
      //添加或者修改监控事件
      void UpdateEvent(Channel* channel)
      {
            bool ret=HasChannel(channel);
            if(ret==false)
            {
                  channels_.insert({channel->GetFd(),channel});
                  Update(channel,EPOLL_CTL_ADD);
            }
            else
            {
                  Update(channel,EPOLL_CTL_MOD);
            }
      }
      //移除监控
      void RemoveEvent(Channel* channel)
      {
            bool ret=HasChannel(channel);
            if(ret)     channels_.erase(channel->GetFd());
            Update(channel,EPOLL_CTL_DEL);
      }
      //开始监控，返回活跃连接
      void Poll(std::vector<Channel*>* active)
      {
            //int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);
            int nfds=epoll_wait(epfd_,evs_,MAX_EPOLLEVENTS,-1);
            if(nfds<0)
            {
                  if(errno==EINTR)
                        return;//信号打断
                  ERR_LOG("epoll_wait fail!!");
                  abort();
            }     
            for(int i=0;i<nfds;i++)
            {
                  auto it=channels_.find(evs_[i].data.fd);
                  assert(it!=channels_.end());
                  it->second->SetREvents(evs_[i].events);//设置实际就绪的事件
                  active->push_back(it->second);
            }
      }
private:
      int epfd_;
      struct epoll_event evs_[MAX_EPOLLEVENTS];
      std::unordered_map<int,Channel*> channels_;
};


using TaskFunc=std::function<void()>;
using ReleaseFunc=std::function<void()>;
class TimerTask
{
public:
      TimerTask(uint64_t id,uint32_t timeout,const TaskFunc& task_cb)
            :id_(id),timeout_(timeout),cancel_(false),task_cb_(task_cb)
      {}
      ~TimerTask()
      {
            if(cancel_==false)      task_cb_();
            release_();
      }
public:
      void Cancel()     {cancel_=true;}
      void SetRelease(const ReleaseFunc& cb)      {release_=cb;}  
      uint32_t DelayTime()    {return timeout_;}
private:
      uint64_t id_;                 //定时任务对象的id
      uint32_t timeout_;            //定时任务对象的超时时间
      bool cancel_;                 //false 表示没有被取消，true表示被取消
      TaskFunc task_cb_;            //定时任务要执行的回调任务
      ReleaseFunc release_;           //用于删除tiemrwheel中保存的定时器对象信息
};

class EventLoop;
class TimerWheel
{
      using PtrTask=std::shared_ptr<TimerTask>;
      using WeakTask=std::weak_ptr<TimerTask>;
private:
      void RemoveTimer(uint64_t id)
      {
            auto it=timer_.find(id);
            if(it!=timer_.end())
                  timer_.erase(it);
      }
      static int CreateTimeFd()
      {
            int timefd=timerfd_create(CLOCK_MONOTONIC,0);
            if(timefd<0)
            {
                  ERR_LOG("timerfd_create fail!!");
                  abort();
            }
            //int timerfd_settime(int fd, int flags,const struct itimerspec *new_value,struct itimerspec *old_value);
            struct itimerspec itime;
            itime.it_value.tv_sec=1;
            itime.it_value.tv_nsec=0;//第一次超时时间为一秒后
            itime.it_interval.tv_sec=1;
            itime.it_interval.tv_nsec=0;//第一次超时后，每次超时的时间间隔
            timerfd_settime(timefd,0,&itime,nullptr);
            return timefd;
      }
      int ReadTimeFd()
      {
            uint64_t times;
            //有可能其他描述符的事件处理花费时间比较长，然后在处理定时器描述符事件的时候，有可能就已经超时了很多次
            //read读取到的数据就是从上一次read之后超时的次数
            int n=read(timerfd_,&times,sizeof(uint64_t));
            if(n<0)
            {
                  ERR_LOG("read timer fail!!");
                  abort();
            }
            return times;
      }
public:
      TimerWheel(EventLoop* loop)
            :tick_(0),capacity_(60),wheel_(capacity_),loop_(loop)
            ,timerfd_(CreateTimeFd()),timer_channel_(new Channel(timerfd_,loop))
      {
            timer_channel_->SetReadCallback(std::bind(&TimerWheel::OnTime,this));
            timer_channel_->EnableRead();//启动读事件监控
      }
      void TimerAddInLoop(uint64_t id,uint32_t delay,const TaskFunc& cb)
      {
            PtrTask ptr(new TimerTask(id,delay,cb));
            ptr->SetRelease(std::bind(&TimerWheel::RemoveTimer,this,id));
            int pos=(tick_+delay)%capacity_;
            wheel_[pos].push_back(ptr);
            timer_[id]=WeakTask(ptr);
      }
      //刷新延迟定时任务
      void TimerRefreshInLoop(uint64_t id)
      {
            //通过保存的定时器对象的weak_ptr来构造一个share_ptr 出来，添加到wheel中
            auto it=timer_.find(id);
            if(it==timer_.end())
                  return;
            PtrTask ptr=it->second.lock();
            uint32_t pos=(ptr->DelayTime()+tick_)%capacity_;
            wheel_[pos].push_back(ptr);
      }
      void TimerCancelInLoop(uint64_t id)
      {
            auto it=timer_.find(id);
            if(it==timer_.end())
                  return;
            PtrTask ptr=it->second.lock();
            if(ptr)
                  ptr->Cancel();
      }
      //这个函数每秒钟执行一次，相当于指针向后移动了一步
      void RunTimeTask()
      {
            tick_=(tick_+1)%capacity_;
            wheel_[tick_].clear();
      }
      void OnTime()
      {
            //根据实际超时的次数，执行对应的超时任务
            int times=ReadTimeFd();
            for(int i=0;i<times;i++)
            {
                  RunTimeTask();
            }
      }
public:
      /*定时器有个timer_成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题*/
      /*如果不想加锁，那就把对定时器的所有操作，都放在同一个线程中进行*/
      void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb);
      void TimerRefresh(uint64_t id);
      void TimerCancel(uint64_t id);
      /*这个接口存在线程安全问题，不能被外界调用，只能在模块内，在对应的EventLoop内执行*/
      bool HasTimer(uint64_t id)
      {
            auto it=timer_.find(id);
            if(it==timer_.end())
                  return false;
            return true;
      }
private:
      int tick_;                    //当前的秒钟，走到哪里，释放哪里，就相当于执行哪里的任务
      int capacity_;                //表盘的最大刻度，《==》 延迟的最大时间
      std::vector<std::vector<PtrTask>> wheel_;
      std::unordered_map<uint64_t,WeakTask> timer_;

      EventLoop* loop_;
      int timerfd_;//定时器描述符--可读事件回调就是读取计数器，执行定时任务
      std::unique_ptr<Channel> timer_channel_;
};

class EventLoop
{
public:
      using Functor=std::function<void()>;
public:
      EventLoop()
            :thread_id_(std::this_thread::get_id())
            ,event_fd_(CreateEventFd())
            ,event_channel_(new Channel(event_fd_,this))
            ,timer_wheel_(this)
      {
            //给eventfd添加可读事件回调函数，读取eventfd事件通知次数
            event_channel_->SetReadCallback(std::bind(&EventLoop::ReadEventFd,this));
            //启动eventfd的读事件监控
            event_channel_->EnableRead();
      }
      //三步走：1.事件监控 2.就绪事件处理 3.执行任务
      void Start()
      {
            while(1)
            {
                  //1.事件监控
                  std::vector<Channel*> active;
                  poller_.Poll(&active);
                  //2.事件处理
                  for(auto& f:active)
                  {
                        f->HandleEvent();
                  }
                  //3.执行任务
                  RunAllTask();
            }
      }
      //判断当前线程是否是EventLoop所处的线程
      bool IsInLoop()
      {
            return thread_id_==std::this_thread::get_id();
      }
      void AssertInLoop()
      {
            assert(thread_id_==std::this_thread::get_id());
      }
      //判断当前执行的任务是否处于当前线程中，如果处于，就执行，否则压入任务队列中
      void RunInLoop(const Functor& cb)
      {
            if(IsInLoop())    cb();
            QueueInLoop(cb);
      }
      //将操作压入任务队列中
      void QueueInLoop(const Functor& cb)
      {
            {
                  std::unique_lock<std::mutex> lock(mutex_);
                  tasks_.push_back(cb);
            }
            //唤醒有可能没有事件就绪而导致的epoll阻塞
            //其实就是给eventfd写入一个数据，eventfd就会触发读事件
            WeakUpEventFd();
      }
      //添加/修改描述符的事件监控
      void UpdateEvent(Channel* channel)
      {
            poller_.UpdateEvent(channel);
      }
      //移除描述符的事件监控
      void RemoveEvent(Channel* channel)
      {
            poller_.RemoveEvent(channel);
      }
      void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb){timer_wheel_.TimerAdd(id,delay,cb);}
      void TimerRefresh(uint64_t id){timer_wheel_.TimerRefresh(id);}
      void TimerCancel(uint64_t id){timer_wheel_.TimerCancel(id);}
      bool HasTimer(uint64_t id){return timer_wheel_.HasTimer(id);}
public:
      //执行任务池的所有任务
      void RunAllTask()
      {
            std::vector<Functor> functor;
            {
                  std::unique_lock<std::mutex> lock(mutex_);
                  tasks_.swap(functor);
            }
            for(auto& f:functor)
                  f();
      }
      static int CreateEventFd()
      {
            int efd=eventfd(0,EFD_CLOEXEC|EFD_NONBLOCK);
            if(efd<0)
            {
                  ERR_LOG("eventfd fail!!");
                  abort();
            }
            return efd;
      }
      void ReadEventFd()
      {
            uint64_t res;
            int n=read(event_fd_,&res,sizeof(res));
            if(n<0)
            {
                  if(errno==EAGAIN||errno==EINTR)
                  {
                        return;
                  }
                  ERR_LOG("readEventFd error");
                  abort();
            }
      }
      void WeakUpEventFd()
      {
            uint64_t val=1;
            int n=write(event_fd_,&val,sizeof(val));
            if(n<0)
            {
                  if(errno==EAGAIN||errno==EINTR)
                  {
                        return;
                  }
                  ERR_LOG("WeakUpEventFd fail!!");
                  abort();
            }
            return;
      }
private:
      std::thread::id thread_id_;               //线程id
      int event_fd_;                            //eventfd唤醒IO事件监控有可能导致的阻塞 
      std::unique_ptr<Channel> event_channel_;
      Poller poller_;                           //进行所有描述符的事件监控
      std::vector<Functor> tasks_;              //任务池
      std::mutex mutex_;                        //实现任务池操作的线程安全
      TimerWheel timer_wheel_;                  //定时器模块
};

class LoopThread
{
public:
      /*创建线程，设置线程入口函数*/
      LoopThread():loop_(nullptr),thread_(std::thread(&LoopThread::ThreadEntry,this)){}
      /*返回当前线程的关联的EventLoop对象指针*/
      EventLoop* GetLoop()
      {
            EventLoop* loop=nullptr;
            {
                  std::unique_lock<std::mutex> lock(mutex_);
                  cond_.wait(lock,[&](){return loop_!=nullptr;});//loop_为nullptr就一直阻塞
                  loop=loop_;
            }
            return loop;
      }
private:
      /*实例化EventLoop对象，唤醒cond_有可能阻塞的线程，并且开始运行EventLoop模块的功能*/
      void ThreadEntry()
      {
            EventLoop loop;
            {
                  std::unique_lock<std::mutex> lock(mutex_);//加锁
                  loop_=&loop;
                  cond_.notify_all();
            }
            loop.Start();
      }
private:
      /*用于实现loop_获取的同步关系，避免线程创建了，但是loop_还没有实例化之前去获取loop_*/
      std::mutex mutex_;            //互斥锁
      std::condition_variable cond_;//条件变量
      EventLoop* loop_;             //EventLoop* 指针变量，这个对象需要再线程内实例化
      std::thread thread_;          //EventLoop对应的线程
};

class Any
{
private:
      class hold
      {
      public:
            virtual ~hold(){}
            virtual const std::type_info& type()=0;
            virtual hold* clone()=0;
      };

      template<class T>
      class placehold:public hold
      {
      public:
            placehold(const T& val):val_(val){}
            //获取子类对象保存的类型
            virtual const std::type_info& type(){return typeid(T);}
            virtual hold* clone(){return new placehold(val_);}
      public:
            T val_;
      };
public:
      hold* content_;
public:
      Any():content_(nullptr){}
      template<class T>
      Any(const T& val):content_(new placehold<T>(val)){}
      Any(const Any& other)
      {
            content_=other.content_?other.content_->clone():nullptr;
      }
      ~Any(){delete content_;}
public:
      //返回子类对象保存数据的指针
      template<class T>
      T* get()
      {
            assert(typeid(T)==content_->type());
            return &(((placehold<T>*)content_)->val_);
      }
      //赋值运算符的函数重载
      template<class T>
      Any& operator=(const T& val)
      {
            Any tmp(val);
            std::swap(content_,tmp.content_);
            return *this;
      }
      Any& operator=(const Any& any)
      {
            Any tmp(any);
            std::swap(content_,tmp.content_);
            return *this;
      }
};

class Connection;
//DISCONNECTED--连接关闭状态                                      CONNECTING--连接建立成功，待处理状态
//CONNECTED--连接建立完成，各种设置已经完成，可以通信的状态          DISCONNECTING--待关闭状态
typedef enum
{
      DISCONNECTED,CONNECTING,CONNECTED,DISCONNECTING
}ConnStatu;

using PtrConnection=std::shared_ptr<Connection>;
class Connection:std::enable_shared_from_this<Connection>
{
private:
      /*这四个回调函数，是让服务器模块设置的（其实服务器模块的处理回调也是组件使用者使用的）*/
      using ConnectedCallback=std::function<void(const PtrConnection&)>;
      using MessageCallback=std::function<void(const PtrConnection&,Buffer*)>;
      using ClosedCallback=std::function<void(const PtrConnection&)>;
      using AnyEventCallback=std::function<void(const PtrConnection&)>;
public:
      Connection(EventLoop* loop,uint64_t conn_id,int sockfd)
            :conn_id_(conn_id),sockfd_(sockfd),enable_inactive_release_(false)
            ,loop_(loop),statu_(CONNECTING),socket_(sockfd_),channel_(sockfd_,loop_)
      {
            channel_.SetCloseCallback(std::bind(&Connection::HandleClose,this));
            channel_.SetErrorCallback(std::bind(&Connection::HandleError,this));
            channel_.SetWriteCallback(std::bind(&Connection::HandleWrite,this));
            channel_.SetReadCallback(std::bind(&Connection::HandleRead,this));
            channel_.SetEventCallback(std::bind(&Connection::HandleEvent,this));
      }
      ~Connection()
      {
            DBG_LOG("RELEASE CONNECTION:%p",this);
      }
      //获取管理的文件描述符
      int GetFd() {return sockfd_;}
      //获取连接id
      uint64_t GetId() {return conn_id_;}
      //是否处于CONNECTED状态
      bool Connected()  {return statu_==CONNECTED;}
      //设置上下文，连接建立完成时调用
      void SetContext(const Any& context) {context_=context;}
      //获取上下文，返回的是指针
      Any* GetContext() {return &context_;}
      void SetConnectedCallback(const ConnectedCallback& cb) {connected_callback_=cb;}
      void SetMessageCallback(const MessageCallback& cb) {message_callback_=cb;}
      void SetClosedCallback(const ClosedCallback& cb) {close_callback_=cb;}
      void SetAnyEventCallback(const AnyEventCallback& cb) {event_callback_=cb;}
      void SetSrvClosedCallback(const ClosedCallback& cb) {server_closed_callback_=cb;}
      //连接建立就绪后，进行channel回调设置，启动读监控，调用connected_callback_
      void Established()
      {
            loop_->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
      }
      //发送数据，将数据放到发送缓冲区，启动写事件监控
      void Send(const char* data,size_t len)
      {
            //外界传入的data,可能是个临时的空间，现在只是把发送操作压入了任务池，有可能并没有被执行
            //因此有可能执行的时候，data指向的空间已经释放了
            Buffer buf;
            buf.WriteAndPush(data,len);
            loop_->RunInLoop(std::bind(&Connection::SendInLoop,this,buf));
      }
      //提供给组件使用者的关闭接口--并不实际关闭，需要判断有没有数据待处理
      void Shutdown()  
      {
            loop_->RunInLoop(std::bind(&Connection::ShutDownInLoop,this));
      }
      void Release()
      {
            loop_->RunInLoop(std::bind(&Connection::ReleaseInLoop,this));
      }
      //启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
      void EnableInactiveRelease(int sec)
      {
            loop_->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));
      }
      //取消非活跃销毁
      void CancelInactiveRelease()
      {
            loop_->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop,this));
      }
      //切换协议--重置上下文以及阶段性回调处理函数--而这个接口必须在EventLoop线程中立即执行
      //防备新的事件触发后，处理的时候，切换任务还没有被执行--会导致任务使用原来的协议处理了
      void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg,
                   const ClosedCallback &closed, const AnyEventCallback &event)
      {
            loop_->AssertInLoop();
            loop_->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
      }
private:
      /*5个channel的事件回调函数 */
      /*描述符可读事件触发后调用的函数，接受socket数据放到接受缓冲区中，让后调用message_callback_*/
      void HandleRead()
      {
            //1.接受socket的数据，放到缓冲区
            char buf[65536]={0};
            ssize_t n=socket_.NonBlockRecv(buf,65535);
            if(n<0)
            {
                  //出错了，不能直接关闭连接
                  return ShutDownInLoop();
            }
            //将数据放入输入缓冲区，并且往后偏移
            in_buffer_.WriteAndPush(buf,n);
            if(in_buffer_.ReadAbleSize()>0)
            {
                  //shared_from_this--从当前对象自身获取自身的shared_ptr管理对象
                  return message_callback_(shared_from_this(),&in_buffer_);
            }
      }
      //描述符可写事件触发后调用的函数，将发送缓冲区的数据进行发送
      void HandleWrite()
      {
            ssize_t n=socket_.NonBlockSend(out_buffer_.ReaderPosition(),out_buffer_.ReadAbleSize());
            if(n<0)
            {
                  //发送错误就该关闭连接了
                  if(in_buffer_.ReadAbleSize()>0)
                        message_callback_(shared_from_this(),&in_buffer_);
                  return Release();//这是真正的关闭连接
            }
            out_buffer_.MoveReadOffset(n);//读偏移向后移动
            if(out_buffer_.ReadAbleSize()==0)
            {
                  channel_.DisableWrite();//没有数据待发送了，关闭写事件监控
                  if(statu_==DISCONNECTING)//如果当前是连接待关闭状态，则有数据，发送完数据释放连接，没有则直接释放
                        return Release();
            }
      }
      //描述符触发挂断事件
      void HandleClose()
      {
            /*一旦连接挂断了，套接字就什么都干不了，有数据待处理就处理一下，完毕关闭连接*/
            if(in_buffer_.ReadAbleSize()>0)
                  message_callback_(shared_from_this(),&in_buffer_);
            return Release();
      }
      //描述符触发出错事件
      void HandleError()
      {
            return HandleClose();
      }
      //描述符触发任意事件：1.刷新连接的活跃度--延迟定时销毁任务 2.调用组件使用者的任意事件回调函数
      void HandleEvent()
      {
            if(enable_inactive_release_)   loop_->TimerRefresh(conn_id_);
            if(event_callback_)           event_callback_(shared_from_this());
      }
      //连接获取之后，所处的状态下要进行各种设置（启动读监控，调用回调函数）
      void EstablishedInLoop()
      {
            //1.修改连接状态 2.启动读事件监控 3.调用回调函数
            assert(statu_==CONNECTING);//当前的状态一定是上层的半连接状态
            statu_=CONNECTED;

            channel_.EnableRead();
            if(connected_callback_) connected_callback_(shared_from_this());
      }
      //这个接口才是实际的释放接口
      void ReleaseInLoop()
      {
            //1.修改连接状态，将其置为DISCONNECTED
            statu_=DISCONNECTED;
            //2.移除连接的事件监控
            channel_.Remove();
            //3.关闭描述符
            socket_.Close();
            //4.如果当前定时任务中还有定时销毁任务，则取消任务
            if(loop_->HasTimer(conn_id_)) CancelInactiveReleaseInLoop();
            //5.调用关闭回调函数，避免先移除服务器管理的连接信息导致Connection被释放，再去处理会出错
            if(close_callback_)     close_callback_(shared_from_this());
            //6.移除服务器内部的管理信息
            if(server_closed_callback_)   server_closed_callback_(shared_from_this());
      }
      //这个接口并不是实际的发送接口，而只是把数据放到了发送缓冲区，启动了可写事件监控
      void SendInLoop(Buffer &buf)
      {
            if (statu_ == DISCONNECTED)
                  return;
            out_buffer_.WriteBufferAndPush(buf);
            if (channel_.WriteAble() == false)
            {
                  channel_.EnableWrite();
            }
      }
      //这个关闭操作并非实际的连接释放操作，需要判断有没有数据待处理，待发送
      void ShutDownInLoop()
      {
            statu_=DISCONNECTING;//设置连接为半关闭状态
            if(in_buffer_.ReadAbleSize()>0)
            {
                  if(message_callback_)
                        message_callback_(shared_from_this(),&in_buffer_);
            }
            //要么就是写入数据的时候出错关闭，要么就是没有待发送数据，直接关闭
            if(out_buffer_.ReadAbleSize()>0)
            {
                  if(channel_.WriteAble()==false)
                        channel_.EnableWrite();
            }
            if(out_buffer_.ReadAbleSize()==0)
                  Release();
      }
      //启动非活跃连接超时释放规则
      void EnableInactiveReleaseInLoop(int sec)
      {
            //1.将enable_inactive_release_标志设置为true
            enable_inactive_release_=true;
            //2.如果当前定时销毁任务存在，那么就刷新延迟一下
            if(loop_->HasTimer(conn_id_))
                  return loop_->TimerRefresh(conn_id_);
            //3.如果不存在定时销毁任务，则新增
            loop_->TimerAdd(conn_id_,sec,std::bind(&Connection::Release,this));
      }
      void CancelInactiveReleaseInLoop()
      {
            enable_inactive_release_=false;
            if(loop_->HasTimer(conn_id_))
                  loop_->TimerCancel(conn_id_);
      }
      void UpgradeInLoop(const Any& context,const ConnectedCallback& con,const MessageCallback& meg,\
      const ClosedCallback& closed,const AnyEventCallback& event)
      {
            context_=context;
            connected_callback_=con;
            message_callback_=meg;
            close_callback_=closed;
            event_callback_=event;
      }
private:
      uint64_t conn_id_;                  //连接的唯一id，便于连接的管理与查找,同时也是定时器id，\
                                          必须是唯一的，为了简化操作，conn_id_就作为定时器id
      int sockfd_;                        //连接关联的文件描述符
      bool enable_inactive_release_;       //连接是否启动非活跃连接的标志，默认为false
      EventLoop* loop_;                   //连接所关联的一个EventLoop
      ConnStatu statu_;                   //连接状态
      Socket socket_;                     //套接字管理
      Channel channel_;                   //连接的事件管理
      Buffer in_buffer_;                 //输入缓冲区--存放从socket读取到的数据
      Buffer out_buffer_;                 //输出缓冲区--存放要发送给对端的数据
      Any context_;                       //请求的接受处理上下文

      ConnectedCallback connected_callback_;
      MessageCallback message_callback_;
      ClosedCallback close_callback_;
      AnyEventCallback  event_callback_;
      /*组件内的连接关闭回调--组件设置的，因为服务器组件内会把所有的连接管理起来*/
      /*一旦某个连接要关闭，就应该从管理的地方移除掉自己的信息*/
      ClosedCallback server_closed_callback_; 
};

class Acceptor
{
private:
      using AcceptCallback=std::function<void(int)>;
      /*监听套接字的读事件回调--获取新连接，调用accept_callback_函数进行新连接处理*/
      void HandleRead()
      {
            int newfd=listensock_.Accept();
            if(newfd<0)       return;
            if(accept_callback_)    accept_callback_(newfd);
      }
      int CreateServer(uint16_t port)
      {
            bool ret=listensock_.CreateServer(port);
            assert(ret==true);
            return listensock_.GetFd();
      }
public:
      /*不能将启动读事件监控，放到构造函数内，必须放到构造函数后*/
      /*否则有可能造成事件启动后，立刻就有文件描述符就绪，处理的时候，回调函数还没有设置；新连接得不到处理，造成资源泄漏*/
      Acceptor(EventLoop *loop,uint16_t port)
            :listensock_(CreateServer(port)),loop_(loop)
            ,channel_(listensock_.GetFd(),loop_)
      {
            channel_.SetReadCallback(std::bind(&Acceptor::HandleRead,this));
      }
      void SetAcceptCallback(const AcceptCallback& cb) {accept_callback_=cb;}
      void Listen()     {channel_.EnableRead();}
private:
      Socket listensock_;     //用于创建监听套接字
      EventLoop *loop_;       //用于对监听套接字进行事件监控
      Channel channel_;        //用于对监听套接字进行事件管理
      AcceptCallback accept_callback_;
};

void Channel::Remove(){loop_->RemoveEvent(this);}
void Channel::Update(){loop_->UpdateEvent(this);}

void TimerWheel::TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb)
{
      loop_->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop,this,id,delay,cb));
}
void TimerWheel::TimerRefresh(uint64_t id)
{
      loop_->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,id));
}
void TimerWheel::TimerCancel(uint64_t id)
{
      loop_->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop,this,id));
}